package shuai7boy.vip.transform.mr.au;

import java.util.List;

import org.apache.commons.configuration.ConfigurationFactory;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.MultipleColumnPrefixFilter;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

import com.google.common.collect.Lists;

import shuai7boy.vip.common.EventLogConstants;
import shuai7boy.vip.common.GlobalConstants;
import shuai7boy.vip.common.KpiTypeEnum;
import shuai7boy.vip.transform.mr.TransformerBaseRunner;
import shuai7boy.vip.transform.mr.TransformerOutputFormat;
import shuai7boy.vip.transformer.model.dim.StatsUserDimension;
import shuai7boy.vip.transformer.model.value.map.TimeOutValue;
import shuai7boy.vip.transformer.model.value.reduce.MapWritableValue;
import shuai7boy.vip.util.TimeUtil;

public class ActiveUserRunner extends TransformerBaseRunner {

	private static final Logger logger = Logger.getLogger(ActiveUserRunner.class);

	public static void main(String[] args) {
		ActiveUserRunner au = new ActiveUserRunner();

		try {
			au.setupRunner("active-user", TransformerBaseRunner.class, ActiveUserMapper.class, ActiveUserReducer.class,
					StatsUserDimension.class, TimeOutValue.class);
			au.startRunner(args);

		} catch (Throwable tb) {
			logger.error("运行active user出错");
			throw new RuntimeException(tb);
		}

	}

	@Override
	protected void configure(String... resourceFiles) {
		super.configure(resourceFiles);
		conf.set("mapred.child.java.opts", "-Xmx500m");
		conf.set("mapreduce.map.output.compress", "true");
	}

	@Override
	public void beforeRun(Job job) {
		logger.info("开始运行。。。");
		job.setNumReduceTasks(3);
		job.setPartitionerClass(PartitionerClass.class);
		job.setReduceSpeculativeExecution(false);
		job.setSpeculativeExecution(false);
	}

	@Override
	protected Filter fetchHbaseFilter() {

		FilterList flist = new FilterList();
		String[] strs = new String[] { EventLogConstants.LOG_COLUMN_NAME_PLATFORM,
				EventLogConstants.LOG_COLUMN_NAME_SERVER_TIME, EventLogConstants.LOG_COLUMN_NAME_BROWSER_NAME,
				EventLogConstants.LOG_COLUMN_NAME_BROWSER_VERSION

		};
		flist.addFilter(super.getColumnFilter(strs));

		return flist;
	}

	/**
	 * 瀵规暟鎹繘琛屽垎鍓�
	 * 
	 * @author RYJ
	 *
	 */
	public class PartitionerClass extends Partitioner<StatsUserDimension, TimeOutValue> {

		@Override
		public int getPartition(StatsUserDimension key, TimeOutValue value, int numPartitions) {

			String kpiName = key.getStatsCommon().getKpi().getKpiName();
			if (KpiTypeEnum.ACTIVE_USER.name.equals(kpiName)) {
				return 0;
			} else if (KpiTypeEnum.BROWSER_ACTIVE_USER.name.equals(kpiName)) {
				return 1;
			} else if (KpiTypeEnum.HOURLY_ACTIVE_USER.name.equals(kpiName)) {
				return 2;
			}

			throw new IllegalArgumentException(
					"鏃犳硶鑾峰彇鍒嗗尯锛屽綋鍓峩pi:" + key.getStatsCommon().getKpi().getKpiName() + ",褰撳墠reduce涓暟:" + numPartitions);
		}

	}

}
